/*
 * JDBCMessageArchiveRepository.java
 *
 * Tigase Jabber/XMPP Server
 * Copyright (C) 2004-2016 "Tigase, Inc." <office@tigase.com>
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License,
 * or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. Look for COPYING file in the top folder.
 * If not, see http://www.gnu.org/licenses/.
 *
 */
package tigase.archive.db;

//~--- non-JDK imports --------------------------------------------------------

import tigase.archive.AbstractCriteria;
import tigase.archive.db.JDBCMessageArchiveRepository.Criteria;
import tigase.db.*;
import tigase.db.DataRepository.dbTypes;
import tigase.util.Base64;
import tigase.xml.DomBuilderHandler;
import tigase.xml.Element;
import tigase.xml.SimpleParser;
import tigase.xml.SingletonFactory;
import tigase.xmpp.BareJID;
import tigase.xmpp.JID;
import tigase.xmpp.RSM;

import java.sql.*;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.*;
import java.util.Date;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Class description
 *
 *
 * @version        Enter version here..., 13/02/16
 * @author         Enter your name here...
 */
@Repository.Meta( supportedUris = { "jdbc:[^:]+:.*" } )
public class JDBCMessageArchiveRepository extends AbstractMessageArchiveRepository<Criteria> {
	public static final String JIDS_ID  = "jid_id";
	public static final String JIDS_JID = "jid";
	public static final String JIDS_DOMAIN = "domain";

	// jids table
	public static final String JIDS_TABLE = "tig_ma_jids";
	private static final Logger log        =
		Logger.getLogger(JDBCMessageArchiveRepository.class.getCanonicalName());
	private static final long LONG_NULL              = 0;
	private static final long MILIS_PER_DAY          = 24 * 60 * 60 * 1000;
	
	public static final String MSGS_ID	        = "msg_id";
	public static final String MSGS_BUDDY_ID  = "buddy_id";
	public static final String MSGS_BUDDY_RESOURCE = "buddy_res";
	private static final String MSGS_BODY      = "body";
	private static final String MSGS_DIRECTION = "direction";
	public static final String MSGS_MSG       = "msg";
	public static final String MSGS_OWNER_ID  = "owner_id";
	protected static final String MSGS_HASH	   = "stanza_hash";

//+ "create unique index " + JIDS_TABLE + "_" + JIDS_JID + " on "
//+ JIDS_TABLE + " ( " + JIDS_JID + "(765));";
	// messages table
	public static final String MSGS_TABLE        = "tig_ma_msgs";
	private static final String MSGS_TIMESTAMP    = "ts";
	private static final String MSGS_TYPE         = "type";
	
	private static final String TAGS_TABLE        = "tig_ma_tags";
	private static final String TAGS_ID           = "tag_id";
	private static final String TAGS_TAG          = "tag";
	private static final String TAGS_OWNER_ID      = "owner_id";
	
	private static final String MSGS_TAGS_TABLE   = "tig_ma_msgs_tags";
	
	private static final SimpleParser parser      = SingletonFactory.getParserInstance();
	private static final String GET_JID_IDS_QUERY = "select " + JIDS_JID + ", " + JIDS_ID +
																									" from " + JIDS_TABLE + " where " +
																									JIDS_JID + " = ?" + " or " + JIDS_JID +
																									" = ?";
	private static final String GET_JID_ID_QUERY = "select " + JIDS_JID + ", " + JIDS_ID +
																								 " from " + JIDS_TABLE + " where " +
																								 JIDS_JID + " = ?";
	private static final String ADD_JID_QUERY = "insert into " + JIDS_TABLE + " (" +
																							JIDS_JID + ", " + JIDS_DOMAIN + ") values (?, ?)";
	private static final String DERBY_CREATE_JIDS = "create table " + JIDS_TABLE + " ( "
																									+ JIDS_ID + " bigint generated by default as identity not null, " + JIDS_JID
																									+ " varchar(2049), primary key ( " + JIDS_ID + " ));"
																									+ "create unique index " + JIDS_TABLE + "_" + JIDS_JID + " on " + JIDS_TABLE
																									+ " (" + JIDS_JID + ");";
	private static final String PGSQL_CREATE_JIDS = "create table " + JIDS_TABLE + " ( "
																									+ JIDS_ID + " bigserial, " + JIDS_JID
																									+ " varchar(2049), primary key ("
																									+ JIDS_ID + ")); "
																									+ "create unique index " + JIDS_TABLE
																									+ "_" + JIDS_JID + " on " + JIDS_TABLE
																									+ " ( " + JIDS_JID + ");";
	private static final String SQLSERVER_CREATE_JIDS = "create table " + JIDS_TABLE + " ( "
																									+ JIDS_ID + " bigint identity(1,1), " + JIDS_JID
																									+ " nvarchar(2049),"
																									+ JIDS_JID + "_fragment as left (" + JIDS_JID + ", 765),"
																									+ "primary key (" + JIDS_ID + ")); "
																									+ "create unique index " + JIDS_TABLE
																									+ "_" + JIDS_JID + " on " + JIDS_TABLE
																									+ " ( " + JIDS_JID + "_fragment " + ");";
	private static final String MYSQL_CREATE_JIDS =
															"create table " + JIDS_TABLE + " ( " + JIDS_ID
															+ " bigint unsigned NOT NULL auto_increment, " + JIDS_JID
															+ " varchar(2049), primary key (" + JIDS_ID + ")); ";

	private static final String[] GET_MESSAGE_FIELDS = { MSGS_MSG, MSGS_TIMESTAMP, MSGS_DIRECTION };
	private static String getMessagesSelectQuery(dbTypes dbType, List<String> fields, String where) {
		StringBuilder sb = new StringBuilder("select ");
		
		int idx = fields.indexOf(JIDS_JID);
		
		switch (dbType) {
			case sqlserver:
			case jtds:
				addFieldsToMessagesStartBuilder(sb, "x", fields);
				sb.append(" from (select ");
				List<String> subqueryFields = new ArrayList<String>(fields);
				if (idx > -1) {
					subqueryFields.remove(idx);
					subqueryFields.add(idx, MSGS_BUDDY_ID);
				}
				addFieldsToMessagesStartBuilder(sb, "m", subqueryFields);
				sb.append(", ROW_NUMBER() over (order by m." + MSGS_TIMESTAMP + ") as rn");
				break;
			default:
				addFieldsToMessagesStartBuilder(sb, "m", fields);
				break;
		}
		sb.append(" from " + MSGS_TABLE + " m");
		switch (dbType) {
			case sqlserver:
			case jtds:
				break;
			default:
				if (idx > -1)
					sb.append(" inner join " + JIDS_TABLE + " b ON b." + JIDS_ID + " = m." + MSGS_BUDDY_ID);
				break;
		}
		sb.append(" where m." + MSGS_OWNER_ID + " = ? ");
		
		sb.append(where);
		
		switch (dbType) {
			case derby:
				sb.append(GENERIC_GET_MESSAGES_ORDER_BY).append(DERBY_LIMIT);
				break;
			case jtds:
			case sqlserver:
				sb.append(") x");
				if (idx > -1) {
					sb.append(" inner join " + JIDS_TABLE + " b ON b." + JIDS_ID + " = x." + MSGS_BUDDY_ID);
				}
				sb.append(MSSQL2008_LIMIT).append(GENERIC_GET_MESSAGES_ORDER_BY);
				break;
			default:
				sb.append(GENERIC_GET_MESSAGES_ORDER_BY).append(GENERIC_LIMIT);
				break;
		}		
		
		return sb.toString();
	}
	
	private static void addFieldsToMessagesStartBuilder(StringBuilder sb, String tablePrefix, List<String> fields) {
		for (int i=0; i<fields.size(); i++) {
			String field = fields.get(i);
			if (i > 0)
				sb.append(", "); 
			if (JIDS_JID.equals(field))
				sb.append("b.").append(JIDS_JID);
			else
				sb.append(tablePrefix).append(".").append(field);
		}		
	}
	
	private static String getMessagesCountQuery(dbTypes dbType, String where) {
		return GENERIC_GET_MESSAGES_COUNT + where;
	}
	
	private static final String GENERIC_GET_MESSAGES_COUNT = "select count(m." + MSGS_TIMESTAMP + ") from " + MSGS_TABLE + " m where m." + MSGS_OWNER_ID 
			+ " = ?";
	private static final String GENERIC_GET_MESSAGES_ORDER_BY = " order by " + MSGS_TIMESTAMP;
	private static final String GENERIC_GET_COLLECTIONS_SELECT = "select min(m." + MSGS_TIMESTAMP + ") as ts, j." + JIDS_JID + " from " + MSGS_TABLE + " m "
			+ "inner join " + JIDS_TABLE + " j on m." + MSGS_BUDDY_ID + " = j." + JIDS_ID + " where m." + MSGS_OWNER_ID + " = ? ";
	private static final String GENERIC_GET_COLLECTIONS_SELECT_WITH_TYPE = "select min(m." + MSGS_TIMESTAMP + ") as ts, j." + JIDS_JID + ", case when m." + MSGS_TYPE + " = 'groupchat' then 'groupchat' else '' end as " + MSGS_TYPE + " from " + MSGS_TABLE + " m "
			+ "inner join " + JIDS_TABLE + " j on m." + MSGS_BUDDY_ID + " = j." + JIDS_ID + " where m." + MSGS_OWNER_ID + " = ? ";
	private static final String MSSQL2008_GET_COLLECTIONS_SELECT = "select x." + MSGS_TIMESTAMP + ", x." + JIDS_JID + " from ( "
			+ "select min(m." + MSGS_TIMESTAMP + ") as " + MSGS_TIMESTAMP + ", j." + JIDS_JID + ", ROW_NUMBER() over (order by min(m." + MSGS_TIMESTAMP + "), j." + JIDS_JID + ") as rn from " + MSGS_TABLE + " m "
			+ "inner join " + JIDS_TABLE + " j on m." + MSGS_BUDDY_ID + " = j." + JIDS_ID + " where m." + MSGS_OWNER_ID + " = ? ";
	private static final String MSSQL2008_GET_COLLECTIONS_SELECT_WITH_TYPE = "select x." + MSGS_TIMESTAMP + ", x." + JIDS_JID + ", x." + MSGS_TYPE + " from ( "
			+ "select min(m." + MSGS_TIMESTAMP + ") as " + MSGS_TIMESTAMP + ", j." + JIDS_JID + ", case when m." + MSGS_TYPE + " = 'groupchat' then 'groupchat' else '' end as " + MSGS_TYPE + ", ROW_NUMBER() over (order by min(m." + MSGS_TIMESTAMP + "), j." + JIDS_JID + ") as rn from " + MSGS_TABLE + " m "
			+ "inner join " + JIDS_TABLE + " j on m." + MSGS_BUDDY_ID + " = j." + JIDS_ID + " where m." + MSGS_OWNER_ID + " = ? ";
	private static final String GENERIC_GET_COLLECTIONS_SELECT_GROUP = " group by date(m." + MSGS_TIMESTAMP + "), m." + MSGS_BUDDY_ID + ", j." + JIDS_JID;
	private static final String GENERIC_GET_COLLECTIONS_SELECT_GROUP_WITH_TYPE = " group by date(m." + MSGS_TIMESTAMP + "), m." + MSGS_BUDDY_ID + ", j." + JIDS_JID + ", case when m." + MSGS_TYPE + " = 'groupchat' then 'groupchat' else '' end ";
	private static final String MSSQL2008_GET_COLLECTIONS_SELECT_GROUP = " group by cast(m." + MSGS_TIMESTAMP + " as date), m." + MSGS_BUDDY_ID + ", j." + JIDS_JID + ") x ";
	private static final String MSSQL2008_GET_COLLECTIONS_SELECT_GROUP_WITH_TYPE = " group by cast(m." + MSGS_TIMESTAMP + " as date), m." + MSGS_BUDDY_ID + ", j." + JIDS_JID + ", case when m." + MSGS_TYPE + " = 'groupchat' then 'groupchat' else '' end " + ") x ";
	private static final String GENERIC_GET_COLLECTIONS_SELECT_ORDER = " order by min(m." + MSGS_TIMESTAMP + "), j." + JIDS_JID;
	private static final String MSSQL2008_GET_COLLECTIONS_SELECT_ORDER = " order by x." + MSGS_TIMESTAMP + ", x." + JIDS_JID;
	private static final String GENERIC_GET_COLLECTIONS_COUNT = "select count(1) from (select min(m." + MSGS_TIMESTAMP + ") as " + MSGS_TIMESTAMP + ", m." + MSGS_BUDDY_ID + " from " 
			+ MSGS_TABLE + " m where m." + MSGS_OWNER_ID + " = ? ";
	private static final String GENERIC_GET_COLLECTIONS_COUNT_GROUP = "group by date(m." + MSGS_TIMESTAMP + "), m." + MSGS_BUDDY_ID + ") x";
	private static final String GENERIC_GET_COLLECTIONS_COUNT_GROUP_WITH_TYPE = "group by date(m." + MSGS_TIMESTAMP + "), m." + MSGS_BUDDY_ID+ ", case when m." + MSGS_TYPE + " = 'groupchat' then 'groupchat' else '' end " + ") x";
	private static final String MSSQL2008_GET_COLLECTIONS_COUNT_GROUP = "group by cast(m." + MSGS_TIMESTAMP + " as date), m." + MSGS_BUDDY_ID + ") x";
	private static final String MSSQL2008_GET_COLLECTIONS_COUNT_GROUP_WITH_TYPE = "group by cast(m." + MSGS_TIMESTAMP + " as date), m." + MSGS_BUDDY_ID+ ", case when m." + MSGS_TYPE + " = 'groupchat' then 'groupchat' else '' end " + ") x";
	private static final String GENERIC_LIMIT = " limit ? offset ?";					// limit, offset
	private static final String DERBY_LIMIT = " offset ? rows fetch next ? rows only";	// offset, limit
	private static final String MSSQL2008_LIMIT = " where x.rn > ? and x.rn <= ?";				// offset, limit + offset

	private static final String DERBY_CREATE_TAGS = "create table " + TAGS_TABLE + " ( "
																									+ TAGS_ID + " bigint generated by default as identity not null, " + TAGS_TAG
																									+ " varchar(255), " + TAGS_OWNER_ID + " bigint references  " + JIDS_TABLE + "(" + JIDS_ID + ") on delete cascade, " 
																									+ "primary key ( " + TAGS_ID + " ));"
																									+ "create index " + TAGS_TABLE + "_" + TAGS_OWNER_ID + " on " + TAGS_TABLE
																									+ " (" + TAGS_OWNER_ID + ");"
																									+ "create unique index " + TAGS_TABLE
																									+ "_" + TAGS_TAG + "_" + TAGS_OWNER_ID + " on " + TAGS_TABLE
																									+ " ( " + TAGS_OWNER_ID + "," + TAGS_TAG + ");";
	private static final String PGSQL_CREATE_TAGS = "create table " + TAGS_TABLE + " ( "
																									+ TAGS_ID + " bigserial, " + TAGS_TAG
																									+ " varchar(255), " + TAGS_OWNER_ID + " bigint not null,"
																									+ "	primary key (" + TAGS_ID + "),"
			+ " foreign key (" + TAGS_OWNER_ID + ") references " + JIDS_TABLE + " (" + JIDS_ID + ") on delete cascade ); "
																									+ "create index " + TAGS_TABLE + "_" + TAGS_OWNER_ID + " on " + TAGS_TABLE
																									+ " (" + TAGS_OWNER_ID + ");"
																									+ "create unique index " + TAGS_TABLE
																									+ "_" + TAGS_TAG + "_" + TAGS_OWNER_ID + " on " + TAGS_TABLE
																									+ " ( " + TAGS_OWNER_ID + "," + TAGS_TAG + ");";
	private static final String SQLSERVER_CREATE_TAGS = "create table " + TAGS_TABLE + " ( "
																									+ TAGS_ID + " bigint identity(1,1), " + TAGS_TAG
																									+ " nvarchar(255),"
			+ TAGS_OWNER_ID + " bigint not null,"
																									+ "primary key (" + TAGS_ID + ")," 
			+ " foreign key (" + TAGS_OWNER_ID + ") references " + JIDS_TABLE + "(" + JIDS_ID + ") on delete cascade ); "
																									+ "create index " + TAGS_TABLE + "_" + TAGS_OWNER_ID + " on " + TAGS_TABLE
																									+ " (" + TAGS_OWNER_ID + ");"
																									+ "create unique index " + TAGS_TABLE
																									+ "_" + TAGS_TAG + "_" + TAGS_OWNER_ID + " on " + TAGS_TABLE
																									+ " ( " + TAGS_OWNER_ID + "," + TAGS_TAG + ");";
	private static final String MYSQL_CREATE_TAGS =
															"create table " + TAGS_TABLE + " ( " + TAGS_ID
															+ " bigint unsigned NOT NULL auto_increment, " + TAGS_TAG
															+ " varchar(255), " + TAGS_OWNER_ID + " bigint unsigned not null, " 
															+ " primary key (" + TAGS_ID + "), foreign key (" + TAGS_OWNER_ID + ") references " + JIDS_TABLE + "(" + JIDS_ID + ") on delete cascade,"
															+ " key " + TAGS_TABLE + "_" + TAGS_OWNER_ID + " ( " + TAGS_OWNER_ID + " ), "
															+ " unique key " + TAGS_TABLE + "_" + TAGS_TAG + "_" + TAGS_OWNER_ID + " ( " + TAGS_OWNER_ID + "," + TAGS_TAG + " )"
															+ " ); ";	
	
	private static final String DERBY_CREATE_MSGS_TAGS = "create table " + MSGS_TAGS_TABLE + " ("
			+ MSGS_ID + " bigint not null references " + MSGS_TABLE + "(" + MSGS_ID + ") on delete cascade, " 
			+ TAGS_ID + " bigint not null references " + TAGS_TABLE + "(" + TAGS_ID + ") on delete cascade);"
			+ "create index " + MSGS_TAGS_TABLE + "_" + MSGS_ID + " on " + MSGS_TAGS_TABLE + " (" + MSGS_ID + ");"
			+ "create index " + MSGS_TAGS_TABLE + "_" + TAGS_ID + " on " + MSGS_TAGS_TABLE + " (" + TAGS_ID + ");";
	private static final String PGSQL_CREATE_MSGS_TAGS = "create table " + MSGS_TAGS_TABLE + " ("
			+ MSGS_ID + " bigint not null, " 
			+ TAGS_ID + " bigint not null, "
			+ "foreign key (" + MSGS_ID + ") references " + MSGS_TABLE + "(" + MSGS_ID + ") on delete cascade, "
			+ "foreign key (" + TAGS_ID + ") references " + TAGS_TABLE + "(" + TAGS_ID + ") on delete cascade);"
			+ "create index " + MSGS_TAGS_TABLE + "_" + MSGS_ID + " on " + MSGS_TAGS_TABLE + " (" + MSGS_ID + ");"
			+ "create index " + MSGS_TAGS_TABLE + "_" + TAGS_ID + " on " + MSGS_TAGS_TABLE + " (" + TAGS_ID + ");";	
	private static final String SQLSERVER_CREATE_MSGS_TAGS = "create table " + MSGS_TAGS_TABLE + " ("
			+ MSGS_ID + " bigint not null, " 
			+ TAGS_ID + " bigint not null, "
			+ "foreign key (" + MSGS_ID + ") references " + MSGS_TABLE + "(" + MSGS_ID + ") on delete cascade, "
			+ "foreign key (" + TAGS_ID + ") references " + TAGS_TABLE + "(" + TAGS_ID + ") on delete cascade);"
			+ "create index " + MSGS_TAGS_TABLE + "_" + MSGS_ID + " on " + MSGS_TAGS_TABLE + " (" + MSGS_ID + ");"
			+ "create index " + MSGS_TAGS_TABLE + "_" + TAGS_ID + " on " + MSGS_TAGS_TABLE + " (" + TAGS_ID + ");";		
	private static final String MYSQL_CREATE_MSGS_TAGS = "create table " + MSGS_TAGS_TABLE + " ("
			+ MSGS_ID + " bigint unsigned not null, " 
			+ TAGS_ID + " bigint unsigned not null, "
			+ "foreign key (" + MSGS_ID + ") references " + MSGS_TABLE + "(" + MSGS_ID + ") on delete cascade, "
			+ "foreign key (" + TAGS_ID + ") references " + TAGS_TABLE + "(" + TAGS_ID + ") on delete cascade, "
			+ " key " + MSGS_TAGS_TABLE + "_" + MSGS_ID + " (" + MSGS_ID + "), "
			+ " key " + MSGS_TAGS_TABLE + "_" + TAGS_ID + " (" + TAGS_ID + ") );";		
	
	private static final String[][] GET_COLLECTIONS_WHERES = { 
			{ "FROM", " and m." + MSGS_TIMESTAMP + " >= ? " },
			{ "TO", " and m." + MSGS_TIMESTAMP + " <= ? " },
			{ "WITH", " and m." + MSGS_BUDDY_ID + " = ? " }
	};
	private static final String[] GET_COLLECTIONS_COMBINATIONS = {
		"", "FROM", "FROM_TO", "FROM_TO_WITH", "FROM_WITH",
		"TO", "TO_WITH", "WITH"
	};
	private static final String ADD_MESSAGE_1 = "insert into " + MSGS_TABLE + " (" +
																						MSGS_OWNER_ID + ", " + MSGS_BUDDY_ID + ", " +
																						MSGS_TIMESTAMP + ", " + MSGS_DIRECTION +
																						", " + MSGS_TYPE + ", " + MSGS_BODY + ", " + MSGS_MSG +
																						", " + MSGS_HASH + ")" +
																						" values (?, ?, ?, ?, ?, ?, ?, ?)";
	private static final String DERBY_ADD_MESSAGE = "insert into " + MSGS_TABLE + " (" +
																						MSGS_OWNER_ID + ", " + MSGS_BUDDY_ID + ", " + MSGS_BUDDY_RESOURCE + ", " +
																						MSGS_TIMESTAMP + ", " + MSGS_DIRECTION +
																						", " + MSGS_TYPE + ", " + MSGS_BODY + ", " + MSGS_MSG +
																						", " + MSGS_HASH + ")" +
																						" select ?, ?, ?, ?, ?, ?, ?, ?, ?" +
																						" from SYSIBM.SYSDUMMY1 " +
																						" where not exists (select 1 from " + MSGS_TABLE + " m where" +
																						" m." + MSGS_OWNER_ID + " = ? and m." + MSGS_BUDDY_ID + " = ? and" +
																						" m." + MSGS_TIMESTAMP + " between ? and ? and m." + MSGS_HASH + " = ? )";
	private static final String PGSQL_ADD_MESSAGE = "insert into " + MSGS_TABLE + " (" +
																						MSGS_OWNER_ID + ", " + MSGS_BUDDY_ID + ", " + MSGS_BUDDY_RESOURCE + ", " +
																						MSGS_TIMESTAMP + ", " + MSGS_DIRECTION +
																						", " + MSGS_TYPE + ", " + MSGS_BODY + ", " + MSGS_MSG +
																						", " + MSGS_HASH + ")" +
																						" select ?, ?, ?, ?, ?, ?, ?, ?, ?" +
																						//" from " + MSGS_TABLE +
																						" where not exists (select 1 from " + MSGS_TABLE + " m where" + 
																						" m." + MSGS_OWNER_ID + " = ? and m." + MSGS_BUDDY_ID + " = ? and" +
																						" m." + MSGS_TIMESTAMP + " between ? and ? and m." + MSGS_HASH + " = ? )";
	private static final String MYSQL_ADD_MESSAGE = "insert into " + MSGS_TABLE + " (" +
																						MSGS_OWNER_ID + ", " + MSGS_BUDDY_ID + ", " + MSGS_BUDDY_RESOURCE + ", " +
																						MSGS_TIMESTAMP + ", " + MSGS_DIRECTION +
																						", " + MSGS_TYPE + ", " + MSGS_BODY + ", " + MSGS_MSG +
																						", " + MSGS_HASH + ") values (?, ?, ? , ?, ?, ?, ?, ?, ?) on duplicate key update " + MSGS_DIRECTION + " = " + MSGS_DIRECTION;	
	private static final String SQLSERVER_ADD_MESSAGE = "insert into " + MSGS_TABLE + " (" +
																						MSGS_OWNER_ID + ", " + MSGS_BUDDY_ID + ", " + MSGS_BUDDY_RESOURCE + ", " +
																						MSGS_TIMESTAMP + ", " + MSGS_DIRECTION +
																						", " + MSGS_TYPE + ", " + MSGS_BODY + ", " + MSGS_MSG +
																						", " + MSGS_HASH + ")" +
																						" select tmp." + MSGS_OWNER_ID + ", tmp." + MSGS_BUDDY_ID + ", tmp." +
																						MSGS_BUDDY_RESOURCE + ", tmp." +
																						MSGS_TIMESTAMP + ", tmp." + MSGS_DIRECTION +
																						", tmp." + MSGS_TYPE + ", tmp." + MSGS_BODY + ", " + MSGS_MSG +
																						", tmp." + MSGS_HASH + " from (select ? as " + MSGS_OWNER_ID + 
																						", ? as " + MSGS_BUDDY_ID + ", ? as " + MSGS_BUDDY_RESOURCE + 
																						", ? as " + MSGS_TIMESTAMP + 
																						", ? as " + MSGS_DIRECTION + ", ? as " + MSGS_TYPE + 
																						", ? as " + MSGS_BODY + ", ? as " + MSGS_MSG + ", ? as " + MSGS_HASH + ") as tmp" +
																						//" from " + MSGS_TABLE +
																						" where not exists (select 1 from " + MSGS_TABLE + " m where" + 
																						" m." + MSGS_OWNER_ID + " = ? and m." + MSGS_BUDDY_ID + " = ? and" +
																						" m." + MSGS_TIMESTAMP + " between ? and ? and m." + MSGS_HASH + " = ?)";

	private static final String DELETE_EXPIRED_MSGS = "delete from " + MSGS_TABLE + " where " + MSGS_TIMESTAMP + " < ? and EXISTS( select 1 from " + JIDS_TABLE + " j " +
																						"where j." + JIDS_ID + " = " + MSGS_OWNER_ID + " and j." + JIDS_DOMAIN + " = ? ) ";
	private static final String REMOVE_MSGS = "delete from " + MSGS_TABLE + " where " +
																						MSGS_OWNER_ID + " = ? and " + MSGS_BUDDY_ID +
																						" = ?" + " and " + MSGS_TIMESTAMP +
																						" <= ? and " + MSGS_TIMESTAMP + " >= ?";
	private static final String DERBY_CREATE_MSGS = "create table " + MSGS_TABLE + " ("
																									+ MSGS_ID + " bigint generated by default as identity not null PRIMARY KEY,"
																									+ MSGS_OWNER_ID + " bigint references " + JIDS_TABLE + "(" + JIDS_ID + "),"
																									+ MSGS_BUDDY_ID + " bigint references " + JIDS_TABLE + "(" + JIDS_ID + "),"
																									+ MSGS_TIMESTAMP + " timestamp, "
																									+ MSGS_DIRECTION + " smallint, "
																									+ MSGS_TYPE + " varchar(20), "
																									+ MSGS_BODY + " varchar(32672), "
																									+ MSGS_MSG + " varchar(32672),"
																									+ MSGS_HASH + " varchar(50));"
																									+ "create index " + MSGS_TABLE + "_" + MSGS_OWNER_ID + "_index on " + MSGS_TABLE
																									+ " (" + MSGS_OWNER_ID + ");"
																									+ "create index " + MSGS_TABLE + "_" + MSGS_OWNER_ID + "_" + MSGS_BUDDY_ID
																									+ "_index on " + MSGS_TABLE + " (" + MSGS_OWNER_ID + ", " + MSGS_BUDDY_ID + ");"
																									+ "create index " + MSGS_TABLE + "_" + MSGS_OWNER_ID + "_" + MSGS_TIMESTAMP + "_"
																									+ MSGS_BUDDY_ID + "_index on " + MSGS_TABLE + " (" + MSGS_OWNER_ID + ", "
																									+ MSGS_TIMESTAMP + ", " + MSGS_BUDDY_ID + ");"
																									+ "create unique index " + MSGS_TABLE + "_" + MSGS_OWNER_ID + "_" + MSGS_TIMESTAMP + "_"
																									+ MSGS_BUDDY_ID + "_" + MSGS_HASH + "_index on " + MSGS_TABLE + " (" + MSGS_OWNER_ID + ", "
																									+ MSGS_TIMESTAMP + ", " + MSGS_BUDDY_ID + ", " + MSGS_HASH + ");";
	private static final String PGSQL_CREATE_MSGS = "create table " + MSGS_TABLE + " (" +
																									MSGS_ID + " bigserial, " +
																									MSGS_OWNER_ID + " bigint, " +
																									MSGS_BUDDY_ID + " bigint, " +
																									MSGS_TIMESTAMP + " timestamp, " +
																									MSGS_DIRECTION + " smallint, " +
																									MSGS_TYPE + " varchar(20), " +
																									MSGS_BODY + " text, " +
																									MSGS_MSG + " text," +
																									MSGS_HASH + " varchar(50)," +
																									" primary key (" + MSGS_ID + ")," +
																									" foreign key (" + MSGS_BUDDY_ID +
																									") references " + JIDS_TABLE + " (" +
																									JIDS_ID + ")," + " foreign key (" +
																									MSGS_OWNER_ID + ") references " +
																									JIDS_TABLE + " (" + JIDS_ID + ") ); " +
																									"create index " + MSGS_TABLE + "_" +
																									MSGS_OWNER_ID + "_index on " +
																									MSGS_TABLE + " ( " + MSGS_OWNER_ID +
																									"); " + "create index " + MSGS_TABLE +
																									"_" + MSGS_OWNER_ID + "_" +
																									MSGS_BUDDY_ID + "_index on " +
																									MSGS_TABLE + " ( " + MSGS_OWNER_ID +
																									", " + MSGS_BUDDY_ID + "); " +
																									"create index " + MSGS_TABLE + "_" +
																									MSGS_OWNER_ID + "_" + MSGS_TIMESTAMP +
																									"_" + MSGS_BUDDY_ID + "_index on " +
																									MSGS_TABLE + " ( " + MSGS_OWNER_ID +
																									", " + MSGS_TIMESTAMP + ", " + 
																									MSGS_BUDDY_ID + "); " +
																									"create unique index " + MSGS_TABLE + "_" +
																									MSGS_OWNER_ID + "_" + MSGS_TIMESTAMP +
																									"_" + MSGS_BUDDY_ID + "_" + MSGS_HASH + "_index on " +
																									MSGS_TABLE + " ( " + MSGS_OWNER_ID +
																									", " + MSGS_TIMESTAMP + ", " + 
																									MSGS_HASH + ", " + MSGS_BUDDY_ID + "); ";
	private static final String SQLSERVER_CREATE_MSGS = "create table " + MSGS_TABLE + " (" +
																									MSGS_ID + " bigint IDENTITY(1,1) NOT NULL, " +
																									MSGS_OWNER_ID + " bigint, " +
																									MSGS_BUDDY_ID + " bigint, " +
																									MSGS_TIMESTAMP + " datetime, " +
																									MSGS_DIRECTION + " smallint, " +
																									MSGS_TYPE + " nvarchar(20)," +
																									MSGS_BODY + " ntext, " +
																									MSGS_MSG + " ntext," +
																									MSGS_HASH + " varchar(50), " + 
																									" primary key (" + MSGS_ID + ")," +
																									" foreign key (" + MSGS_BUDDY_ID +
																									") references " + JIDS_TABLE + " (" +
																									JIDS_ID + ")," + " foreign key (" +
																									MSGS_OWNER_ID + ") references " +
																									JIDS_TABLE + " (" + JIDS_ID + ") ); " +
																									"create index " + MSGS_TABLE + "_" +
																									MSGS_OWNER_ID + "_index on " +
																									MSGS_TABLE + " ( " + MSGS_OWNER_ID +
																									"); " + "create index " + MSGS_TABLE +
																									"_" + MSGS_OWNER_ID + "_" +
																									MSGS_BUDDY_ID + "_index on " +
																									MSGS_TABLE + " ( " + MSGS_OWNER_ID +
																									", " + MSGS_BUDDY_ID + "); " +
																									"create index " + MSGS_TABLE + "_" +
																									MSGS_OWNER_ID + "_" + MSGS_TIMESTAMP +
																									"_" + MSGS_BUDDY_ID + "_index on " +
																									MSGS_TABLE + " ( " + MSGS_OWNER_ID +
																									", " + MSGS_TIMESTAMP + ", " +
																									MSGS_BUDDY_ID + "); " + 
																									"create unique index " + MSGS_TABLE + "_" +
																									MSGS_OWNER_ID + "_" + MSGS_TIMESTAMP + "_" +
																									MSGS_BUDDY_ID + "_" + MSGS_HASH + "_index on " + 
																									MSGS_TABLE + " (" + 
																									MSGS_OWNER_ID + ", " + MSGS_TIMESTAMP + ", " +
																									MSGS_BUDDY_ID + ", " + MSGS_HASH + ");";
	private static final String MYSQL_CREATE_MSGS = "create table " + MSGS_TABLE + " (" +
																									MSGS_ID + " bigint unsigned NOT NULL auto_increment, " + 
																									MSGS_OWNER_ID + " bigint unsigned, " +
																									MSGS_BUDDY_ID + " bigint unsigned, " +
																									MSGS_TIMESTAMP + " timestamp, " +
																									MSGS_DIRECTION + " smallint, " +
																									MSGS_TYPE + " varchar(20)," +
																									MSGS_BODY + " text, " + 
																									MSGS_MSG + " text," +
																									MSGS_HASH + " varchar(50), " +
																									" primary key (" + MSGS_ID + "), " +
																									" foreign key (" + MSGS_BUDDY_ID +
																									") references " + JIDS_TABLE + " (" +
																									JIDS_ID + ")," + " foreign key (" +
																									MSGS_OWNER_ID + ") references " +
																									JIDS_TABLE + " (" + JIDS_ID + "), " +
																									"key (" + MSGS_OWNER_ID + "), " + 
																									"key (" + MSGS_OWNER_ID + ", " + 
																									MSGS_BUDDY_ID + "), key (" + 
																									MSGS_OWNER_ID + ", " + MSGS_TIMESTAMP + 
																									", " + MSGS_BUDDY_ID + ")," +
																									" unique index using hash (" +
																									MSGS_OWNER_ID + ", " + MSGS_TIMESTAMP + ", " +
																									MSGS_BUDDY_ID + ", " + MSGS_HASH +	"));";

	private static final String ADD_TAG = "insert into " + TAGS_TABLE + " (" + TAGS_OWNER_ID + ", " + TAGS_TAG + ") values (?,?)";
	private static final String ADD_MESSAGE_TAG = "insert into " + MSGS_TAGS_TABLE + " (" + MSGS_ID + ", " + TAGS_ID + ") values (?,?)";
	private static final String GET_TAG_IDS = "select " + TAGS_ID + ", " + TAGS_TAG + " from " + TAGS_TABLE + " WHERE " + TAGS_OWNER_ID + " = ? AND ( ";
	private static final String GET_TAG_IDS_WHERE_PART = TAGS_TAG + " = ? ";
	private static final String GET_TAG_IDS_END = " )";
	
	private static final String GET_TAGS_FOR_USER = "select t." + TAGS_TAG + " from " + TAGS_TABLE + " t inner join " + JIDS_TABLE 
			+ " j on t." + TAGS_OWNER_ID + " = j." + JIDS_ID + " where j." + JIDS_JID + " = ? and t." + TAGS_TAG + " like ? ";
	private static final String GET_TAGS_FOR_USER_COUNT = "select count(t." + TAGS_ID + ") from " + TAGS_TABLE + " t inner join " + JIDS_TABLE 
			+ " j on t." + TAGS_OWNER_ID + " = j." + JIDS_ID + " where j." + JIDS_JID + " = ? and t." + TAGS_TAG + " like ? ";
	private static final String GET_TAGS_FOR_USER_ORDER = " order by " + TAGS_TAG;
	private static final String SQLSERVER_GET_TAGS_FOR_USER = "select x." + TAGS_TAG + " from ("
			+ " select t." + TAGS_TAG + ", ROW_NUMBER() over (order by t." + TAGS_TAG + ") as rn from " + TAGS_TABLE + " t inner join " + JIDS_TABLE 
			+ " j on t." + TAGS_OWNER_ID + " = j." + JIDS_ID + " where j." + JIDS_JID + " = ? and t." + TAGS_TAG + " like ? ) x ";
	
	private static final String STORE_PLAINTEXT_BODY_KEY = "store-plaintext-body";
	private static final String GROUP_BY_TYPE_KEY = "group-by-chat-type";
	
	private static final String DELETE_EXPIRED_QUERY_TIMEOUT_KEY = "remove-expired-messages-query-timeout";
	private static final int DEF_DELETE_EXPIRED_QUERY_TIMEOUT_VAL = 5 * 60;
	
	//~--- fields ---------------------------------------------------------------

	protected String ADD_MESSAGE;
	
	protected DataRepository data_repo = null;
	private boolean storePlaintextBody = true;
	private boolean groupByType = false;
	private int delete_expired_timeout = DEF_DELETE_EXPIRED_QUERY_TIMEOUT_VAL;

	protected String[] getCollectionsCombinations() {
		return GET_COLLECTIONS_COMBINATIONS;
	}
	
	protected String[][] getCollectionsWheres() {
		return GET_COLLECTIONS_WHERES;
	}
	
	//~--- methods --------------------------------------------------------------

	/**
	 * Method description
	 *
	 *
	 * @param conn_str
	 * @param params
	 *
	 * @throws SQLException
	 */
	@Override
	public void initRepository(String conn_str, Map<String, String> params)
					throws DBInitException {
		try {
			data_repo = RepositoryFactory.getDataRepository( null, conn_str, params );
			if (params.containsKey(STORE_PLAINTEXT_BODY_KEY)) {
				storePlaintextBody = Boolean.parseBoolean(params.get(STORE_PLAINTEXT_BODY_KEY));
			} else {
				storePlaintextBody = true;
			}

			initRepositoryDbSchema();

			switch ( data_repo.getDatabaseType() ) {
				case derby:
					ADD_MESSAGE = DERBY_ADD_MESSAGE;
					break;
				case postgresql:
					ADD_MESSAGE = PGSQL_ADD_MESSAGE;
					break;
				case mysql:
					ADD_MESSAGE = MYSQL_ADD_MESSAGE;
					break;
				case jtds:
				case sqlserver:
					ADD_MESSAGE = SQLSERVER_ADD_MESSAGE;
					break;
			}
			
			// this parameter is set by plugins as we need to initialize statements
			// using config from component and not from processors
			if (!params.containsKey("ignoreStatementInitialization")) {
				StringBuilder sb = new StringBuilder();
				for (Map.Entry<String, String> e : params.entrySet()) {
					sb.append(", " + e.getKey() + " = " + e.getValue());
				}

				if (params.containsKey(GROUP_BY_TYPE_KEY)) {
					groupByType = Boolean.parseBoolean(params.get(GROUP_BY_TYPE_KEY));
				} else {
					groupByType = false;
				}
			}
			
			if (params.containsKey(DELETE_EXPIRED_QUERY_TIMEOUT_KEY)) {
				delete_expired_timeout = Integer.parseInt(params.get(DELETE_EXPIRED_QUERY_TIMEOUT_KEY));
				log.log(Level.FINEST, "setting " + DELETE_EXPIRED_QUERY_TIMEOUT_KEY + " to {0}", delete_expired_timeout);
			}
			
			initPreparedStatements(params);
		} catch (Exception ex) {
			log.log(Level.WARNING, "MessageArchiveDB initialization exception", ex);
		}
	}

	protected void initRepositoryDbSchema() throws SQLException {
		// create tables if not exist
		switch (data_repo.getDatabaseType()) {
			case mysql:
				data_repo.checkTable(JIDS_TABLE, MYSQL_CREATE_JIDS);
				data_repo.checkTable(MSGS_TABLE, MYSQL_CREATE_MSGS);
				break;
			case derby:
				data_repo.checkTable(JIDS_TABLE, DERBY_CREATE_JIDS);
				data_repo.checkTable(MSGS_TABLE, DERBY_CREATE_MSGS);
				break;
			case postgresql:
				data_repo.checkTable(JIDS_TABLE, PGSQL_CREATE_JIDS);
				data_repo.checkTable(MSGS_TABLE, PGSQL_CREATE_MSGS);
				break;
			case jtds:
			case sqlserver:
				data_repo.checkTable(JIDS_TABLE, SQLSERVER_CREATE_JIDS);
				data_repo.checkTable(MSGS_TABLE, SQLSERVER_CREATE_MSGS);
				break;
		}

		checkDB();

		switch (data_repo.getDatabaseType()) {
			case mysql:
				data_repo.checkTable(TAGS_TABLE, MYSQL_CREATE_TAGS);
				data_repo.checkTable(MSGS_TAGS_TABLE, MYSQL_CREATE_MSGS_TAGS);
				break;
			case derby:
				data_repo.checkTable(TAGS_TABLE, DERBY_CREATE_TAGS);
				data_repo.checkTable(MSGS_TAGS_TABLE, DERBY_CREATE_MSGS_TAGS);
				break;
			case postgresql:
				data_repo.checkTable(TAGS_TABLE, PGSQL_CREATE_TAGS);
				data_repo.checkTable(MSGS_TAGS_TABLE, PGSQL_CREATE_MSGS_TAGS);
				break;
			case jtds:
			case sqlserver:
				data_repo.checkTable(TAGS_TABLE, SQLSERVER_CREATE_TAGS);
				data_repo.checkTable(MSGS_TAGS_TABLE, SQLSERVER_CREATE_MSGS_TAGS);
				break;
		}
	}
	
	protected void initPreparedStatements(Map<String,String> params) throws SQLException {
		if (params.containsKey("ignoreStatementInitialization"))
			return;
		data_repo.initPreparedStatement(ADD_JID_QUERY, ADD_JID_QUERY);
		data_repo.initPreparedStatement(GET_JID_ID_QUERY, GET_JID_ID_QUERY);
		data_repo.initPreparedStatement(GET_JID_IDS_QUERY, GET_JID_IDS_QUERY);
		data_repo.initPreparedStatement(ADD_MESSAGE, ADD_MESSAGE, Statement.RETURN_GENERATED_KEYS);
		data_repo.initPreparedStatement(DELETE_EXPIRED_MSGS, DELETE_EXPIRED_MSGS);

			//data_repo.initPreparedStatement(GET_COLLECTIONS, GET_COLLECTIONS);
		Map<String, String> combinations = new HashMap<String, String>();
		for (String combination : getCollectionsCombinations()) {
			StringBuilder sbMain = new StringBuilder();

			if (!combination.isEmpty()) {
				String[] whereParts = combination.split("_");

				for (String part : whereParts) {
					for (String[] where : getCollectionsWheres()) {
						if (!part.equals(where[0])) {
							continue;
						}

						sbMain.append(where[1]);
					}
				}
			}

			for (int j = 0; j < 6; j++) {
				StringBuilder combinationSb1 = new StringBuilder().append(combination);
				StringBuilder querySb1 = new StringBuilder().append(sbMain);

				if (j > 0) {
					if (combinationSb1.length() > 0) {
						combinationSb1.append("_");
					}
					combinationSb1.append("TAGS[").append(j).append("]");
					querySb1.append(" and EXISTS( select 1 from ").append(TAGS_TABLE)
							.append(" t inner join ").append(MSGS_TAGS_TABLE).append(" tm on t.")
							.append(TAGS_ID).append(" = tm.").append(TAGS_ID).append(" where m.")
							.append(MSGS_ID).append(" = tm.").append(MSGS_ID).append(" and (");
					for (int x = 0; x < j; x++) {
						if (x > 0) {
							querySb1.append(" or ");
						}
						querySb1.append("t.").append(TAGS_TAG).append(" = ?");
					}
					querySb1.append(" )) ");
				}

				for (int i = 0; i < 6; i++) {
					StringBuilder combinationSb = new StringBuilder().append(combinationSb1);
					StringBuilder querySb = new StringBuilder().append(querySb1);

					if (i > 0) {
						if (combinationSb.length() > 0) {
							combinationSb.append("_");
						}
						combinationSb.append("CONTAINS[").append(i).append("]");
						for (int x = 0; x < i; x++) {
							querySb.append(" and m.").append(MSGS_BODY).append(" like ? ");
						}
					}

					combinations.put(combinationSb.toString(), querySb.toString());
				}
			}
		}

		for (Map.Entry<String, String> entry : combinations.entrySet()) {
			StringBuilder select = new StringBuilder();
			StringBuilder count = new StringBuilder().append(GENERIC_GET_COLLECTIONS_COUNT);

			switch (data_repo.getDatabaseType()) {
				case jtds:
				case sqlserver:
					if (groupByType)
						select.append(MSSQL2008_GET_COLLECTIONS_SELECT_WITH_TYPE);
					else
						select.append(MSSQL2008_GET_COLLECTIONS_SELECT);
					break;
				default:
					if (groupByType)
						select.append(GENERIC_GET_COLLECTIONS_SELECT_WITH_TYPE);
					else
						select.append(GENERIC_GET_COLLECTIONS_SELECT);
					break;
			}

			select.append(entry.getValue());
			count.append(entry.getValue());

			switch (data_repo.getDatabaseType()) {
				case jtds:
				case sqlserver:
					if (groupByType) {
						select.append(MSSQL2008_GET_COLLECTIONS_SELECT_GROUP_WITH_TYPE);
						count.append(MSSQL2008_GET_COLLECTIONS_COUNT_GROUP_WITH_TYPE);
						
					} else {
						select.append(MSSQL2008_GET_COLLECTIONS_SELECT_GROUP);
						count.append(MSSQL2008_GET_COLLECTIONS_COUNT_GROUP);
					}
					break;
				default:
					if (groupByType) {
						select.append(GENERIC_GET_COLLECTIONS_SELECT_GROUP_WITH_TYPE + GENERIC_GET_COLLECTIONS_SELECT_ORDER);
						count.append(GENERIC_GET_COLLECTIONS_COUNT_GROUP_WITH_TYPE);
					} else {
						select.append(GENERIC_GET_COLLECTIONS_SELECT_GROUP + GENERIC_GET_COLLECTIONS_SELECT_ORDER);
						count.append(GENERIC_GET_COLLECTIONS_COUNT_GROUP);
					}
					break;
			}

			switch (data_repo.getDatabaseType()) {
				case derby:
					select.append(DERBY_LIMIT);
					break;
				case jtds:
				case sqlserver:
					select.append(MSSQL2008_LIMIT).append(MSSQL2008_GET_COLLECTIONS_SELECT_ORDER);
					break;
				default:
					select.append(GENERIC_LIMIT);
					break;
			}

			if (log.isLoggable(Level.FINEST)) {
				log.log(Level.FINEST, "prepared collection select query for " + entry.getKey() + " as '" + select.toString() + "'");
				log.log(Level.FINEST, "prepared collection count query for " + entry.getKey() + " as '" + count.toString() + "'");
			}

			data_repo.initPreparedStatement("GET_COLLECTIONS_" + entry.getKey() + "_SELECT", select.toString());
			data_repo.initPreparedStatement("GET_COLLECTIONS_" + entry.getKey() + "_COUNT", count.toString());
		}

		for (Map.Entry<String, String> entry : combinations.entrySet()) {
			
			List<String> fields = getMessageFields(entry.getKey());

			String select = getMessagesSelectQuery(data_repo.getDatabaseType(), fields, entry.getValue());
			String count = getMessagesCountQuery(data_repo.getDatabaseType(), entry.getValue());

			if (log.isLoggable(Level.FINEST)) {
				log.log(Level.FINEST, "prepared messages select query for " + entry.getKey() + " as '" + select + "'");
				log.log(Level.FINEST, "prepared messages count query for " + entry.getKey() + " as '" + count + "'");
			}

			data_repo.initPreparedStatement("GET_MESSAGES_" + entry.getKey() + "_SELECT", select);
			data_repo.initPreparedStatement("GET_MESSAGES_" + entry.getKey() + "_COUNT", count);
		}

		data_repo.initPreparedStatement(REMOVE_MSGS, REMOVE_MSGS);

		for (int i = 0; i <= 5; i++) {
			StringBuilder select = new StringBuilder().append(GET_TAG_IDS);
			for (int j = 1; j <= i; j++) {
				if (j > 1) {
					select.append(" or ");
				}
				select.append(GET_TAG_IDS_WHERE_PART);
			}
			if (i == 0) {
				select.append("1=1");
			}
			select.append(")");
			data_repo.initPreparedStatement(GET_TAG_IDS + "_" + i, select.toString());
		}
		data_repo.initPreparedStatement(ADD_TAG, ADD_TAG, Statement.RETURN_GENERATED_KEYS);
		data_repo.initPreparedStatement(ADD_MESSAGE_TAG, ADD_MESSAGE_TAG);

		data_repo.initPreparedStatement(GET_TAGS_FOR_USER_COUNT, GET_TAGS_FOR_USER_COUNT);
		switch (data_repo.getDatabaseType()) {
			case derby:
				data_repo.initPreparedStatement(GET_TAGS_FOR_USER, GET_TAGS_FOR_USER + GET_TAGS_FOR_USER_ORDER + DERBY_LIMIT);
				break;
			case jtds:
			case sqlserver:
				data_repo.initPreparedStatement(GET_TAGS_FOR_USER, SQLSERVER_GET_TAGS_FOR_USER + MSSQL2008_LIMIT + GET_TAGS_FOR_USER_ORDER);
				break;
			default:
				data_repo.initPreparedStatement(GET_TAGS_FOR_USER, GET_TAGS_FOR_USER + GET_TAGS_FOR_USER_ORDER + GENERIC_LIMIT);
				break;
		}		
	}
	
	@Override
	public void destroy() {
		// here we use cached instance of repository pool cached by RepositoryFactory
		// so we should not close it
	}
	
	private void checkDB() {
		Statement stmt = null;
		try {
			try {
				stmt = data_repo.createStatement(null);
				stmt.executeQuery("select " + MSGS_BODY + " from " + MSGS_TABLE + " where " + MSGS_OWNER_ID + " = 0");
			} catch (SQLException ex) {
				// if this happens then we have issue with old database schema and missing body columns in MSGS_TABLE
				String alterTable = null;
				switch (data_repo.getDatabaseType()) {
					case derby:
						alterTable = "alter table " + MSGS_TABLE + " add " + MSGS_BODY + " varchar(32672)";
						break;
					case mysql:
						alterTable = "alter table " + MSGS_TABLE + " add " + MSGS_BODY + " text";
						break;
					case postgresql:
						alterTable = "alter table " + MSGS_TABLE + " add " + MSGS_BODY + " text";
						break;
					case jtds:
					case sqlserver:
						alterTable = "alter table " + MSGS_TABLE + " add " + MSGS_BODY + " ntext";
						break;
				}
				try {
					stmt.execute(alterTable);
				} catch (SQLException ex1) {
					log.log(Level.SEVERE, "could not alter table " + MSGS_TABLE + " to add missing column by SQL:\n" + alterTable, ex1);
				}
			}
			data_repo.release(stmt, null);
			try {
				stmt = data_repo.createStatement(null);
				stmt.executeQuery("select " + MSGS_ID + " from " + MSGS_TABLE + " where " + MSGS_OWNER_ID + " = 0");
			} catch (SQLException ex) {
				// if this happens then we have issue with old database schema and missing id column in MSGS_TABLE
				String alterTable = null;
				try {
					switch (data_repo.getDatabaseType()) {
						case derby:
							alterTable = "alter table " + MSGS_TABLE + " add " + MSGS_ID + " bigint generated by default as identity not null";
							stmt.execute(alterTable);
							alterTable = "alter table " + MSGS_TABLE + " add primary key (" + MSGS_ID + ")";
							stmt.execute(alterTable);
							break;
						case mysql:
							alterTable = "alter table " + MSGS_TABLE + " add " + MSGS_ID + " bigint unsigned NOT NULL auto_increment, add primary key (" + MSGS_ID + ")";
							stmt.execute(alterTable);
							break;
						case postgresql:
							alterTable = "alter table " + MSGS_TABLE + " add " + MSGS_ID + " serial";
							stmt.execute(alterTable);
							alterTable = "alter table " + MSGS_TABLE + " add primary key (" + MSGS_ID + ")";
							stmt.execute(alterTable);
							break;
						case jtds:
						case sqlserver:
							alterTable = "alter table " + MSGS_TABLE + " add " + MSGS_ID + " bigint identity(1,1)";
							stmt.execute(alterTable);
							alterTable = "alter table " + MSGS_TABLE + " add primary key (" + MSGS_ID + ")";
							stmt.execute(alterTable);
							break;
					}
				} catch (SQLException ex1) {
					log.log(Level.SEVERE, "could not alter table " + MSGS_TABLE + " to add missing column by SQL:\n" + alterTable, ex1);
				}
			}		
			data_repo.release(stmt, null);
			try {
				stmt = data_repo.createStatement(null);
				//insert into tig_ma_msgs (type) VALUES ( 'unsubscribed' );
				int executeUpdate = stmt.executeUpdate("insert into " + MSGS_TABLE + " ( " + MSGS_TYPE + " )  VALUES (  \"unsubscribed\"  )", Statement.RETURN_GENERATED_KEYS );
				if ( executeUpdate > 0 ){
					ResultSet rs = stmt.getGeneratedKeys();
					if ( rs.next() ){
						int id = rs.getInt( 1 );
						stmt.executeUpdate( "delete from " + MSGS_TABLE + " where " + MSGS_ID + " = " + id );
					}
				}
			} catch (SQLException ex) {
				// if this happens then we have issue with MSGS_TABLE having to short field for message type! need to increase.
				String alterTable = null;
				log.log( Level.INFO, "altering table " + MSGS_TABLE + " extend size of " + MSGS_TYPE + " column" );
				try {
					switch (data_repo.getDatabaseType()) {
						case derby:
							alterTable = "alter table " + MSGS_TABLE + " ALTER COLUMN " + MSGS_TYPE + " SET DATA TYPE varchar(20)";
							stmt.execute(alterTable);
							break;
						case mysql:
							alterTable = "alter table " + MSGS_TABLE + " MODIFY COLUMN " + MSGS_TYPE + " varchar(20);";
							stmt.execute(alterTable);
							break;
						case postgresql:
							alterTable = "alter table " + MSGS_TABLE + " ALTER COLUMN " + MSGS_TYPE + " TYPE varchar(20);";
							stmt.execute(alterTable);
							break;
						case jtds:
						case sqlserver:
							alterTable = "alter table " + MSGS_TABLE + " ALTER COLUMN " + MSGS_TYPE + " varchar(20);";
							stmt.execute(alterTable);
							break;
					}
				} catch (SQLException ex1) {
					log.log(Level.SEVERE, "could not alter table " + MSGS_TABLE + " to increase " + MSGS_TYPE + " column lenght:\n" + alterTable, ex1);
				}
			}
			data_repo.release(stmt, null);
			try {
				stmt = data_repo.createStatement(null);
				stmt.executeQuery("select " + MSGS_HASH + " from " + MSGS_TABLE + " where " + MSGS_OWNER_ID + " = 0");
			} catch (SQLException ex) {
				// if this happens then we have issue with old database schema and missing hash column in MSGS_TABLE
				String alterTable = null;
				try {
					switch (data_repo.getDatabaseType()) {
						case jtds:
						case sqlserver:
							alterTable = "alter table " + MSGS_TABLE + " add " + MSGS_HASH + " varchar(50)";
							stmt.execute(alterTable);
//							// on SQL Server unique index assumes that two null values are the same!
//							alterTable = "update " + MSGS_TABLE + " set " + MSGS_HASH + " = " + MSGS_ID;
//							stmt.execute(alterTable);
							alterTable = "create unique index " + MSGS_TABLE + "_" + MSGS_OWNER_ID + "_" + MSGS_TIMESTAMP + "_" 
									+ MSGS_BUDDY_ID + "_" + MSGS_HASH + "_index on " + MSGS_TABLE + " (" + MSGS_OWNER_ID + ", "
									+ MSGS_TIMESTAMP + ", " + MSGS_BUDDY_ID + ", " + MSGS_HASH + ") WHERE " + MSGS_HASH + " is not null";
							stmt.execute(alterTable);
							break;
						default:
							alterTable = "alter table " + MSGS_TABLE + " add " + MSGS_HASH + " varchar(50)";
							stmt.execute(alterTable);
							alterTable = "create unique index " + MSGS_TABLE + "_" + MSGS_OWNER_ID + "_" + MSGS_TIMESTAMP + "_" 
									+ MSGS_BUDDY_ID + "_" + MSGS_HASH + "_index on " + MSGS_TABLE + " (" + MSGS_OWNER_ID + ", "
									+ MSGS_TIMESTAMP + ", " + MSGS_BUDDY_ID + ", " + MSGS_HASH + ")";
							stmt.execute(alterTable);
							break;							
					}
				} catch (SQLException ex1) {
					log.log(Level.SEVERE, "could not alter table " + MSGS_TABLE + " to add missing column by SQL:\n" + alterTable, ex1);
				}
			}	
			data_repo.release(stmt, null);
			try {
				stmt = data_repo.createStatement(null);
				stmt.executeQuery("select " + MSGS_BUDDY_RESOURCE + " from " + MSGS_TABLE + " where " + MSGS_OWNER_ID + " = 0");
			} catch (SQLException ex) {
				String alterTable = null;
				try {
					switch (data_repo.getDatabaseType()) {
						case jtds:
						case sqlserver:
							alterTable = "alter table " + MSGS_TABLE + " add " + MSGS_BUDDY_RESOURCE + " nvarchar(1024)";
							break;
						default:
							alterTable = "alter table " + MSGS_TABLE + " add " + MSGS_BUDDY_RESOURCE + " varchar(1024)";
							break;
					}
					stmt.execute(alterTable);
					switch (data_repo.getDatabaseType()) {
						case mysql:
							alterTable = "create index " + MSGS_TABLE + "_" + MSGS_OWNER_ID + "_" + MSGS_BUDDY_ID + "_" + MSGS_BUDDY_RESOURCE
									+ "_index on " + MSGS_TABLE + " ("  + MSGS_OWNER_ID + "," + MSGS_BUDDY_ID + "," + MSGS_BUDDY_RESOURCE + "(255))";
							break;
						case jtds:
						case sqlserver:
							alterTable = null;
							break;
						default:
							alterTable = "create index " + MSGS_TABLE + "_" + MSGS_OWNER_ID + "_" + MSGS_BUDDY_ID + "_" + MSGS_BUDDY_RESOURCE
									+ "_index on " + MSGS_TABLE + " ("  + MSGS_OWNER_ID + "," + MSGS_BUDDY_ID + "," + MSGS_BUDDY_RESOURCE + ")";
							break;
					}
					if (alterTable != null)
						stmt.execute(alterTable);					
				} catch (SQLException ex1) {
					log.log(Level.SEVERE, "could not alter table " + MSGS_TABLE + " to add missing column by SQL:\n" + alterTable, ex1);
				}
			} finally {
				data_repo.release(stmt, null);
			}		
			try {
				stmt = data_repo.createStatement(null);
				stmt.executeQuery("select " + JIDS_DOMAIN + " from " + JIDS_TABLE + " where 1 = 0");					
			} catch (SQLException ex) {
				String alterTable = null;
				try {
					switch (data_repo.getDatabaseType()) {
						case jtds:
						case sqlserver:
							alterTable = "alter table " + JIDS_TABLE + " add " + JIDS_DOMAIN + " nvarchar(1024)";
							break;
						default:
							alterTable = "alter table " + JIDS_TABLE + " add " + JIDS_DOMAIN + " varchar(1024)";
							break;
					}
					stmt.execute(alterTable);		
				} catch (SQLException ex1) {
					log.log(Level.SEVERE, "could not alter table " + JIDS_TABLE + " to add missing column by SQL:\n" + alterTable, ex1);
				}
			} finally {
				data_repo.release(stmt, null);
			}
			try {
				stmt = data_repo.createStatement(null);
				ResultSet rs = stmt.executeQuery("select count(" + JIDS_ID + ") from " + JIDS_TABLE + " where " + JIDS_DOMAIN + " IS NULL");
				rs.next();
				int count = rs.getInt(1);
				rs.close();
				
				if (count > 0) {
					String alterTable = null;
					switch (data_repo.getDatabaseType()) {
						case jtds:
						case sqlserver:
							alterTable = "update " + JIDS_TABLE + " set " + JIDS_DOMAIN + " = SUBSTRING(" + JIDS_JID + ", CHARINDEX('@'," + JIDS_JID + ") + 1, LEN(" + JIDS_JID + ")) WHERE " + JIDS_DOMAIN + " IS NULL";
							break;
						case derby:
						case mysql:
							alterTable = "update " + JIDS_TABLE + " set " + JIDS_DOMAIN + " = SUBSTR(" + JIDS_JID + ", LOCATE('@', " + JIDS_JID + ") + 1) WHERE " + JIDS_DOMAIN + " IS NULL";
							break;
						case postgresql:
							alterTable = "update " + JIDS_TABLE + " set " + JIDS_DOMAIN + " = substr(" + JIDS_JID + ", strpos(" + JIDS_JID + ", '@') + 1) WHERE " + JIDS_DOMAIN + " IS NULL";
							break;
						default:
							break;
					}
					stmt.execute(alterTable);		
				}
			} catch (SQLException ex) {
				log.log(Level.SEVERE, "could not update table " + JIDS_TABLE + " to add missing values for column " + JIDS_DOMAIN + " :\n", ex);
			} finally {
				data_repo.release(stmt, null);
			}
		} finally {
			data_repo.release(stmt, null);
		}
		String alterTable = "";
		try {
			stmt = data_repo.createStatement(null);
			switch (data_repo.getDatabaseType()) {
				case mysql:
					alterTable = "create index " + JIDS_TABLE + "_" + JIDS_DOMAIN
							+ "_index on " + JIDS_TABLE + " (" + JIDS_DOMAIN + "(255))";
					break;
				case jtds:
				case sqlserver:
					alterTable = null;
					break;
				default:
					alterTable = "create index " + JIDS_TABLE + "_" + JIDS_DOMAIN
							+ "_index on " + JIDS_TABLE + " (" + JIDS_DOMAIN + ")";
					break;
			}
			if (alterTable != null) {
				stmt.execute(alterTable);
			}
			alterTable = "create index " + MSGS_TABLE + "_" + MSGS_TIMESTAMP
					+ "_index on " + MSGS_TABLE + " (" + MSGS_TIMESTAMP + ")";
			if (alterTable != null) {
				stmt.execute(alterTable);
			}
			log.log(Level.FINEST, "added missing index timestamp and domain");
		} catch (SQLException ex1) {
			// we ignore this exception as there is no good way to check if index exists
			// so we try to add this indexes every time - so exception is expected
		} finally {
			data_repo.release(stmt, null);
		}
		// convert localpart of JID and domain part to lowercase
		try {
			String query = "select 1 from " + JIDS_TABLE + " where " + JIDS_TABLE + "." + JIDS_JID + " <> LOWER(" + JIDS_TABLE + "." + JIDS_JID +")";
			stmt = data_repo.createStatement(null);
			ResultSet rs = stmt.executeQuery(query);
			if (rs.next()) {
				stmt.executeUpdate("update " + JIDS_TABLE + " set " + JIDS_JID + " = LOWER(" + JIDS_JID + "), "
						+ JIDS_DOMAIN + " = LOWER(" + JIDS_DOMAIN + ") where " + JIDS_JID + " <> LOWER(" + JIDS_JID + ") or " + JIDS_DOMAIN + " <> LOWER(" + JIDS_DOMAIN + ")");
			}
			rs.close();
			log.log(Level.FINEST, "lowercased domain and local part of barejid ");
		} catch (SQLException ex1) {
			log.log(Level.SEVERE, "could not update '" + JIDS_JID  + "' and '" + JIDS_DOMAIN + "' with lowercased values", ex1);
		} finally {
			data_repo.release(stmt, null);
		}

		try {
			if (data_repo.getDatabaseType() == dbTypes.mysql) {
				alterTable = "create unique index " + MSGS_TABLE + "_" + MSGS_OWNER_ID + "_" + MSGS_BUDDY_ID + "_" +
						MSGS_HASH + "_index on " + MSGS_TABLE + " (" + MSGS_OWNER_ID + ", " + MSGS_BUDDY_ID + ", " +
						MSGS_HASH + ")";
				stmt = data_repo.createStatement(null);
				stmt.execute(alterTable);
			}
		} catch (SQLException ex) {
		} finally {
			data_repo.release(stmt, null);
		}
		try {
			if (data_repo.getDatabaseType() == dbTypes.mysql) {
				stmt = data_repo.createStatement(null);
				stmt.execute("create index " + JIDS_TABLE + "_" + JIDS_JID + "_index on " + JIDS_TABLE + " (" + JIDS_JID + "(255))");
			}
		} catch (SQLException ex) {
		} finally {
			data_repo.release(stmt, null);
		}
	}
	
	/**
	 * Method description
	 *
	 *
	 * @param owner
	 * @param buddy
	 * @param direction
	 * @param timestamp
	 * @param msg
	 * @param tags
	 */
	@Override
	public void archiveMessage(BareJID owner, JID buddy, Direction direction, Date timestamp, Element msg, Set<String> tags) {
		archiveMessage(owner, buddy, direction, timestamp, msg, tags, null);
	}
		
	protected void archiveMessage(BareJID owner, JID buddy, Direction direction, Date timestamp, Element msg, Set<String> tags, Map<String,Object> additionalData) {
		try {
			ResultSet rs = null;
			String owner_str         = owner.toString();
			String buddy_str         = buddy.getBareJID().toString();
			long[] jids_ids          = getJidsIds(owner_str, buddy_str);
			long owner_id            = (jids_ids[0] != LONG_NULL)
																 ? jids_ids[0]
																 : addJidId(owner);
			long buddy_id            = (jids_ids[1] != LONG_NULL)
																 ? jids_ids[1]
																 : (buddy_str.equals(owner_str) ? owner_id : addJidId(buddy.getBareJID()));
			java.sql.Timestamp mtime = new java.sql.Timestamp(timestamp.getTime());
			msg.addAttribute("time", String.valueOf(mtime.getTime()));

			String type                      = msg.getAttributeStaticStr("type");
			String msgStr                    = msg.toString();
			String body                      = storePlaintextBody ? msg.getChildCData(MSG_BODY_PATH) : null;
			String hash						 = generateHashOfMessageAsString(direction, msg, mtime, additionalData);
			PreparedStatement add_message_st = data_repo.getPreparedStatement(owner,
																					 ADD_MESSAGE);

			Long msgId = null;
			
			synchronized (add_message_st) {
				try {
					int i = 1;
					add_message_st.setLong(i++, owner_id);
					add_message_st.setLong(i++, buddy_id);
					add_message_st.setString(i++, buddy.getResource());
					add_message_st.setTimestamp(i++, mtime);
					add_message_st.setShort(i++, direction.getValue());
					add_message_st.setString(i++, type);
					add_message_st.setString(i++, body);
					add_message_st.setString(i++, msgStr);
					add_message_st.setString(i++, hash);

					i = addMessageAdditionalInfo(add_message_st, i, additionalData);

					if (data_repo.getDatabaseType() != dbTypes.mysql) {
						add_message_st.setLong(i++, owner_id);
						add_message_st.setLong(i++, buddy_id);
						if ("groupchat".equals(type)) {
							add_message_st.setTimestamp(i++, new java.sql.Timestamp(mtime.getTime() - 30 * 60 * 1000));
							add_message_st.setTimestamp(i++, new java.sql.Timestamp(mtime.getTime() + 30 * 60 * 1000));
						} else {
							add_message_st.setTimestamp(i++, mtime);
							add_message_st.setTimestamp(i++, mtime);
						}
						add_message_st.setString(i++, hash);
					}
					add_message_st.executeUpdate();

					if (tags != null) {
						rs = add_message_st.getGeneratedKeys();
						if (rs.next()) {
							switch (data_repo.getDatabaseType()) {
								case postgresql:
									msgId = rs.getLong(MSGS_ID);
									break;
								default:
									msgId = rs.getLong(1);
									break;
							}
						}
					}
				} finally {
					data_repo.release(null, rs);
				}
			}
			
			// in case we tried to archive message which was already archived (ie. by other 
			// session or cluster node) server may ignore insert so it will not return id of inserted
			// record as insert was not executed 
			// in this case we need to exit from this function
			if (msgId == null)
				return;
			
			if (tags != null && !tags.isEmpty()) {
				Map<String,Long> tagsMap = ensureTags(owner, owner_id, tags);
				PreparedStatement add_message_tag_st = data_repo.getPreparedStatement(owner, ADD_MESSAGE_TAG);
				synchronized (add_message_tag_st) {
					for (Long tagId : tagsMap.values()) {
						add_message_tag_st.setLong(1, msgId);
						add_message_tag_st.setLong(2, tagId);
						add_message_tag_st.addBatch();
					}
					add_message_tag_st.executeBatch();
				}
			}
		} catch (SQLException ex) {
			log.log(Level.WARNING, "Problem adding new entry to DB: " + msg, ex);
		}
	}

	protected int addMessageAdditionalInfo(PreparedStatement stmt, int i, Map<String,Object> additionalData) throws SQLException {
		return i;
	}	
	
	@Override
	public void deleteExpiredMessages(BareJID owner, LocalDateTime before) throws TigaseDBException {
		try {
			PreparedStatement delete_expired_msgs_st = data_repo.getPreparedStatement(owner, DELETE_EXPIRED_MSGS);
			long timestamp_long = before.toEpochSecond(ZoneOffset.UTC) * 1000;
			Timestamp ts = new java.sql.Timestamp(timestamp_long);
			synchronized (delete_expired_msgs_st) {
				if (log.isLoggable(Level.FINEST)) {
					log.log(Level.FINEST, "executing removal of expired messages for domain {0} with timeout set to {1} seconds", 
							new Object[]{owner, delete_expired_timeout});
				}
				delete_expired_msgs_st.setQueryTimeout(delete_expired_timeout);
				delete_expired_msgs_st.setTimestamp(1, ts);
				delete_expired_msgs_st.setString(2, owner.toString().toLowerCase());
				delete_expired_msgs_st.executeUpdate();
			}
		} catch (SQLException ex) {
			throw new TigaseDBException("Could not remove expired messages", ex);
		}
	}
	
	private Map<String,Long> ensureTags(BareJID owner, long owner_id, Set<String> tags) throws SQLException {
		Map<String, Long> tagsMap = new HashMap<String, Long>();
		ResultSet rs = null;
		Iterator<String> it = tags.iterator();
		int iters = (tags.size() / 5) + 1;
		for (int i = 0; i < iters; i++) {
			int params = (i == (iters - 1)) ? tags.size() % 5 : 5;
			PreparedStatement get_tag_ids_st = data_repo.getPreparedStatement(owner, GET_TAG_IDS + "_" + params);
			synchronized (get_tag_ids_st) {
				try {
					get_tag_ids_st.setLong(1, owner_id);
					for (int j = 0; j < params; j++) {
						String tag = it.next();
						get_tag_ids_st.setString(j + 2, tag);
					}
					rs = get_tag_ids_st.executeQuery();
					while (rs.next()) {
						long id = rs.getLong(1);
						String tag = rs.getString(2);
						tagsMap.put(tag, id);
					}
				} finally {
					data_repo.release(null, rs);
				}
				rs = null;
			}
		}

		if (tagsMap.size() < tags.size()) {
			PreparedStatement add_tag_st = data_repo.getPreparedStatement(owner, ADD_TAG);
			for (String tag : tags) {
				if (tagsMap.containsKey(tag))
					continue;

				synchronized (add_tag_st) {
					try {
						add_tag_st.setLong(1, owner_id);
						add_tag_st.setString(2, tag);
						add_tag_st.executeUpdate();
						rs = add_tag_st.getGeneratedKeys();
						if (rs.next()) {
							tagsMap.put(tag, rs.getLong(1));
						}
					} finally {
						data_repo.release(null, rs);
					}

					rs = null;
				}
			}
		}
		
		return tagsMap;
	}
	
	//~--- get methods ----------------------------------------------------------

	/**
	 * Method description
	 *
	 *
	 * @param owner
	 * @param crit
	 *
	 * @return
	 * @throws tigase.db.TigaseDBException
	 */
	@Override
	public List<Element> getCollections(BareJID owner, Criteria crit)
					 throws TigaseDBException {
		try {
			long[] jids_ids = crit.getWith() == null ? getJidsIds(owner.toString()) : getJidsIds(owner.toString(), crit.getWith());

			crit.setOwnerId(jids_ids[0]);
			if (jids_ids.length > 1)
				crit.setBuddyId(jids_ids[1]);

			Integer count = getCollectionsCount(owner, crit);
			if (count == null)
				count = 0;
			crit.setSize(count);

			List<Element> results = getCollectionsItems(owner, crit);

			RSM rsm = crit.getRSM();
			rsm.setResults(count, crit.getOffset());
			if (!results.isEmpty()) {
				rsm.setFirst(String.valueOf(crit.getOffset()));
				rsm.setLast(String.valueOf(crit.getOffset() + (results.size() - 1)));
			}

			return results;
		} catch (SQLException ex) {
			throw new TigaseDBException("Cound not retrieve collections", ex);
		}		
	}

	/**
	 * Method description
	 *
	 *
	 * @param owner
	 * @param crit
	 *
	 * @return
	 * @throws tigase.db.TigaseDBException
	 */
	@Override
	public List<Element> getItems(BareJID owner, Criteria crit)
					 throws TigaseDBException {
		try {
			long[] jids_ids = crit.getWith() != null ? getJidsIds(owner.toString(), crit.getWith()) : getJidsIds(owner.toString());

			crit.setOwnerId(jids_ids[0]);
			if (jids_ids.length > 1)
				crit.setBuddyId(jids_ids[1]);


			Integer count = getItemsCount(owner, crit);
			if (count == null) {
				count = 0;
			}
			crit.setSize(count);

			List<Element> items = getItemsItems(owner, crit);

			RSM rsm = crit.getRSM();
			rsm.setResults(count, crit.getOffset());
			if (items!= null && !items.isEmpty()) {
				rsm.setFirst(String.valueOf(crit.getOffset()));
				rsm.setLast(String.valueOf(crit.getOffset() + (items.size() - 1)));
			}

			return items;
		} catch (SQLException ex) {
			throw new TigaseDBException("Cound not retrieve items", ex);
		}		
	}

	//~--- methods --------------------------------------------------------------

	/**
	 * Method description
	 *
	 *
	 * @param owner
	 * @param withJid
	 * @param start
	 * @param end
	 *
	 * @throws TigaseDBException
	 */
	@Override
	public void removeItems(BareJID owner, String withJid, Date start, Date end)
					throws TigaseDBException {
		try {
			long[] jids_ids = getJidsIds(owner.toString(), withJid);

			if (start == null) {
				start = new Date(0);
			}
			if (end == null) {
				end = new Date(0);
			}

			java.sql.Timestamp start_ = new java.sql.Timestamp(start.getTime());
			java.sql.Timestamp end_ = new java.sql.Timestamp(end.getTime());
			PreparedStatement remove_msgs_st = data_repo.getPreparedStatement(owner, REMOVE_MSGS);

			synchronized (remove_msgs_st) {
				synchronized (remove_msgs_st) {
					remove_msgs_st.setLong(1, jids_ids[0]);
					remove_msgs_st.setLong(2, jids_ids[1]);
					remove_msgs_st.setTimestamp(3, end_);
					remove_msgs_st.setTimestamp(4, start_);
					remove_msgs_st.executeUpdate();
				}
			}
		} catch (SQLException ex) {
			throw new TigaseDBException("Cound not remove items", ex);
		}
	}

	/**
	 * Method description
	 * 
	 * @param owner
	 * @param startsWith
	 * @param crit
	 * @return
	 * @throws TigaseDBException 
	 */
	@Override
	public List<String> getTags(BareJID owner, String startsWith, Criteria crit) throws TigaseDBException {
		List<String> results = new ArrayList<String>();
		try {
			ResultSet rs = null;
			int count = 0;
			startsWith = startsWith + "%";
			
			PreparedStatement get_tags_count_st = data_repo.getPreparedStatement(owner, GET_TAGS_FOR_USER_COUNT);
			synchronized (get_tags_count_st) {
				try {
					get_tags_count_st.setString(1, owner.toString().toLowerCase());
					get_tags_count_st.setString(2, startsWith);

					rs = get_tags_count_st.executeQuery();
					if (rs.next()) {
						count = rs.getInt(1);
					}
				} finally {
					data_repo.release(null, rs);
				}
			}
			crit.setSize(count);

			PreparedStatement get_tags_st = data_repo.getPreparedStatement(owner, GET_TAGS_FOR_USER);
			synchronized (get_tags_st) {
				try {
					int i = 1;
					get_tags_st.setString(i++, owner.toString().toLowerCase());
					get_tags_st.setString(i++, startsWith);

					switch (data_repo.getDatabaseType()) {
						case derby:
							get_tags_st.setInt(i++, crit.getOffset());
							get_tags_st.setInt(i++, crit.getLimit());
							break;
						case jtds:
						case sqlserver:
							get_tags_st.setInt(i++, crit.getOffset());
							get_tags_st.setInt(i++, crit.getOffset() + crit.getLimit());
							break;
						default:
							get_tags_st.setInt(i++, crit.getLimit());
							get_tags_st.setInt(i++, crit.getOffset());
							break;
					}

					rs = get_tags_st.executeQuery();
					while (rs.next()) {
						results.add(rs.getString(1));
					}
				} finally {
					data_repo.release(null, rs);
				}
			}
			
			RSM rsm = crit.getRSM();
			rsm.setResults(count, crit.getOffset());
			if (results!= null && !results.isEmpty()) {
				rsm.setFirst(String.valueOf(crit.getOffset()));
				rsm.setLast(String.valueOf(crit.getOffset() + (results.size() - 1)));
			}			
		} catch (SQLException ex) {
			throw new TigaseDBException("Could not retrieve known tags from database", ex);
		}
		
		return results;
	}

	private List<Element> getCollectionsItems(BareJID owner, Criteria crit)
					throws SQLException {
		List<Element> results = new LinkedList<Element>();
		ResultSet selectRs = null;
		PreparedStatement get_collections_st = data_repo.getPreparedStatement(owner, "GET_COLLECTIONS_"
				+ crit.getQueryName() + "_SELECT");

		int i = 2;
		synchronized (get_collections_st) {
			try {
				crit.setItemsQueryParams(get_collections_st, data_repo.getDatabaseType());

				selectRs = get_collections_st.executeQuery();
				while (selectRs.next()) {
					Timestamp startTs = selectRs.getTimestamp(1);
					String with = selectRs.getString(2);
					String type = null;
					if (groupByType) {
						type = selectRs.getString(3);
					}
					addCollectionToResults(results, crit, with, startTs, type);
				}
			} finally {
				data_repo.release(null, selectRs);
			}
		}
		return results;
	}	
	
	private Integer getCollectionsCount(BareJID owner, Criteria crit) throws SQLException {
		ResultSet countRs = null;
		Integer count = null;
		PreparedStatement get_collections_count = data_repo.getPreparedStatement(owner, "GET_COLLECTIONS_"
				+ crit.getQueryName() + "_COUNT");
		int i = 2;
		synchronized (get_collections_count) {
			try {
				crit.setCountQueryParams(get_collections_count, data_repo.getDatabaseType());
				countRs = get_collections_count.executeQuery();
				if (countRs.next()) {
					count = countRs.getInt(1);
				}
			} finally {
				data_repo.release(null, countRs);
			}
		}
		return count;
	}
	
	private List<Element> getItemsItems(BareJID owner, Criteria crit) throws SQLException {
		ResultSet rs      = null;		
		Queue<Item> results = new ArrayDeque<Item>();
		int i=1;
		boolean containsWith = crit.getQueryName().contains("WITH");
		PreparedStatement get_messages_st = data_repo.getPreparedStatement(owner, "GET_MESSAGES_" + crit.getQueryName() + "_SELECT");
		synchronized (get_messages_st) {
			try {
				crit.setItemsQueryParams(get_messages_st, data_repo.getDatabaseType());

				rs = get_messages_st.executeQuery();
				while (rs.next()) {
					Item item = newItemInstance();
					item.read(rs, crit, containsWith);
					results.offer(item);
				}
			} finally {
				data_repo.release(null, rs);
			}
		}

		List<Element> msgs = new LinkedList<Element>();

		if (!results.isEmpty()) {
			DomBuilderHandler domHandler = new DomBuilderHandler();

			Date startTimestamp = crit.getStart();
			Item item = null;
			while ((item = results.poll()) != null) {
				// workaround for case in which start was not specified
				if (startTimestamp == null)
					startTimestamp = item.timestamp;
				
				parser.parse(domHandler, item.message.toCharArray(), 0, item.message.length());

				Queue<Element> queue = domHandler.getParsedElements();
				Element msg = null;

				while ((msg = queue.poll()) != null) {
					addMessageToResults(msgs, crit, startTimestamp, item, msg);
				}			
			}

			crit.setStart(startTimestamp);
			
			// no point in sorting messages by secs attribute as messages are already
			// sorted in SQL query and also this sorting is incorrect
//			Collections.sort(msgs, new Comparator<Element>() {
//				@Override
//				public int compare(Element m1, Element m2) {
//					return m1.getAttributeStaticStr("secs").compareTo(
//							m2.getAttributeStaticStr("secs"));
//				}
//			});
		}

		return msgs;		
	}
	
	protected Element addMessageToResults(List<Element> msgs, Criteria crit, Date startTimestamp, Item item, Element msg) {
		return addMessageToResults(msgs, crit, startTimestamp, msg, item.timestamp, item.direction, item.with);
	}
	
	protected Item newItemInstance() {
		return new Item();
	}
	
	protected List<String> getMessageFields(String combination) {
		List<String> fields = new ArrayList<String>();
		fields.addAll(Arrays.asList(GET_MESSAGE_FIELDS));
		
		if (!combination.contains("WITH"))
			fields.add(JIDS_JID);
		
		return fields;
	}

	private Integer getItemsCount(BareJID owner, Criteria crit) throws SQLException {
		Integer count = null;
		ResultSet rs = null;
		PreparedStatement get_messages_st = data_repo.getPreparedStatement(owner, "GET_MESSAGES_" + crit.getQueryName() + "_COUNT");
		synchronized (get_messages_st) {
			try {
				crit.setCountQueryParams(get_messages_st, data_repo.getDatabaseType());

				rs = get_messages_st.executeQuery();
				if (rs.next()) {
					count = rs.getInt(1);
				}
			} finally {
				data_repo.release(null, rs);
			}
		}
		return count;
	}
	
	/**
	 * Method description
	 *
	 *
	 * @param jids
	 *
	 * @return
	 *
	 * @throws SQLException
	 */
	protected long[] getJidsIds(String... jids) throws SQLException {
		ResultSet rs = null;
		long[] results = new long[jids.length];

		Arrays.fill(results, LONG_NULL);
		if (jids.length == 1) {
			PreparedStatement get_jid_id_st = data_repo.getPreparedStatement(null,
					GET_JID_ID_QUERY);

			synchronized (get_jid_id_st) {
				try {
					get_jid_id_st.setString(1, jids[0].toLowerCase());
					rs = get_jid_id_st.executeQuery();
					if (rs.next()) {
						results[0] = rs.getLong("jid_id");

						return results;
					}
				} finally {
					data_repo.release(null, rs);
				}
			}

			return results;
		} else {
			PreparedStatement get_jids_id_st = data_repo.getPreparedStatement(null,
					GET_JID_IDS_QUERY);

			synchronized (get_jids_id_st) {
				try {
					for (int i = 0; i < jids.length; i++) {
						get_jids_id_st.setString(i + 1, jids[i].toLowerCase());
					}
					rs = get_jids_id_st.executeQuery();

					int cnt = 0;

					while (rs.next()) {
						String db_jid = rs.getString("jid");

						for (int i = 0; i < jids.length; i++) {
							if (db_jid.equalsIgnoreCase(jids[i])) {
								results[i] = rs.getLong("jid_id");
								++cnt;
							}
						}
					}
				} finally {
					data_repo.release(null, rs);
				}

				return results;
			}
		}
	}

	//~--- methods --------------------------------------------------------------

	private long addJidId(BareJID jid) throws SQLException {
		PreparedStatement add_jid_st = data_repo.getPreparedStatement(null, ADD_JID_QUERY);

		try {
			synchronized (add_jid_st) {
				add_jid_st.setString(1, jid.toString().toLowerCase());
				add_jid_st.setString(2, jid.getDomain().toLowerCase());
				add_jid_st.executeUpdate();
			}
		} catch (SQLException ex) {
			log.log(Level.FINEST, "Exception adding jid to tig_ma_jids table, it may occur "
					+ "if other thread added this jid in the meantime", ex);
		}

		// This is not the most effective solution but this method shouldn't be
		// called very often so the perfrmance impact should be insignificant.
		long[] jid_ids = getJidsIds(jid.toString());

		if (jid_ids != null) {
			return jid_ids[0];
		} else {

			// That should never happen here, but just in case....
			log.log(Level.WARNING, "I have just added new jid but it was not found.... {0}",
							jid);

			return LONG_NULL;
		}
	}
	
	private String generateHashOfMessageAsString(Direction direction, Element msg, Date ts, Map<String,Object> additionalData) {
		byte[] result = generateHashOfMessage(direction, msg, ts, additionalData);
		return result != null ? Base64.encode(result) : null;
	}

	@Override
	public AbstractCriteria newCriteriaInstance() {
		return new Criteria();
	}
	
	public static class Item<Crit extends Criteria> {
		String message;
		Date timestamp;
		Direction direction;
		String with;
		
		protected int read(ResultSet rs, Crit crit, boolean containsWith) throws SQLException {
			int i = 1;
			message = rs.getString(i++);
			timestamp = rs.getTimestamp(i++);
			direction = Direction.getDirection(rs.getShort(i++));
			if (!containsWith) {
				with = rs.getString(i++);
			}
			return i;
		}
	}
	
	public static class Criteria extends AbstractCriteria<Timestamp> {

		private long ownerId;
		private long buddyId;
		protected String queryName;
		
		@Override
		protected Timestamp convertTimestamp(Date date) {
			if (date == null)
				return null;
			return new Timestamp(date.getTime());
		}
		
		public String getQueryName() {
			if (queryName == null)
				return updateQueryName();
			return queryName;
		}
		
		public String updateQueryName() {
			StringBuilder query = new StringBuilder(20);
			if (getStart() != null) {
				query.append("FROM");
			}
			if (getEnd() != null) {
				if (query.length() > 0) {
					query.append("_");
				}
				query.append("TO");
			}
			if (getWith() != null) {
				if (query.length() > 0) {
					query.append("_");
				}
				query.append("WITH");
			} else {
				// not supported
			}
			if (!getTags().isEmpty()) {
				if (query.length() > 0) {
					query.append("_");
				}
				query.append("TAGS[").append(getTags().size()).append("]");
			}
			if (!getContains().isEmpty()) {
				if (query.length() > 0) {
					query.append("_");
				}
				query.append("CONTAINS[").append(getContains().size()).append("]");
			}
			queryName = query.toString();
			return queryName;
		}
		
		public void setOwnerId(Long id) {
			if (id == null)
				ownerId = 0;
			else
				ownerId = id;
		}
		
		public void setBuddyId(Long id) {
			if (id == null)
				buddyId = 0;
			else
				buddyId = id;
		}
		
		public int setCountQueryParams(PreparedStatement stmt, DataRepository.dbTypes dbType) throws SQLException {
			int i=1;
			stmt.setLong(i++, ownerId);
			return setCountQueryParams(stmt, dbType, i);
		}
		
		protected int setCountQueryParams(PreparedStatement stmt, DataRepository.dbTypes dbType, int i) throws SQLException {	
			if (getStart() != null) {
				stmt.setTimestamp(i++, getStart());
			}
			if (getEnd() != null) {
				stmt.setTimestamp(i++, getEnd());
			}
			if (getWith() != null) {
				stmt.setLong(i++, buddyId);
			}
			for (String tag : getTags()) {
				stmt.setString(i++, tag);
			}
			for (String contains : getContains()) {
				stmt.setString(i++, "%" + contains + "%");
			}
			return i;
		}
		
		public void setItemsQueryParams(PreparedStatement stmt, DataRepository.dbTypes dbType) throws SQLException {
			int i = setCountQueryParams(stmt, dbType);
			switch (dbType) {
				case derby:
					stmt.setInt(i++, getOffset());
					stmt.setInt(i++, getLimit());
					break;
				case jtds:
				case sqlserver:
					stmt.setInt(i++, getOffset());
					stmt.setInt(i++, getOffset() + getLimit());
					break;
				default:
					stmt.setInt(i++, getLimit());
					stmt.setInt(i++, getOffset());
					break;
			}
		}

	}
	
}


//~ Formatted in Tigase Code Convention on 13/02/20
